package com.wangxin.study8.threadtest;

import java.util.concurrent.*;

/**
 * @author jojo.wang
 * @title: QueueTest
 * @projectName study8
 * @description: Queue
 * @date 2019/5/617:44
 */

public class QueueTest {
    public static void main(String[] args) {
        /**
         * ArrayBlockingQueue 是 BlockingQueue 接口的有界队列实现类，底层采用数组来实现。
         * ArrayBlockingQueue一旦创建，容量不能改变。
         * 其并发控制采用可重入锁来控制，不管是插入操作还是读取操作，都需要获取到锁才能进行操作。
         * 当队列容量满时，尝试将元素放入队列将导致操作阻塞;尝试从一个空队列中取一个元素也会同样阻塞。
         * ArrayBlockingQueue 默认情况下不能保证线程访问队列的公平性，
         * 所谓公平性是指严格按照线程等待的绝对时间顺序，
         * 即最先等待的线程能够最先访问到 ArrayBlockingQueue。
         * 而非公平性则是指访问 ArrayBlockingQueue 的顺序不是遵守严格的时间顺序，
         * 有可能存在，当 ArrayBlockingQueue 可以被访问时，
         * 长时间阻塞的线程依然无法访问到 ArrayBlockingQueue。如果保证公平性，通常会降低吞吐量。
         */
        ArrayBlockingQueue<Integer> arrayBlockingQueue = new ArrayBlockingQueue(8);
        for (int i = 0; i < 30; i++) {
            Thread thread = new Thread(new ThreadPut(arrayBlockingQueue,"ArrayBlockingQueue",i));
            thread.start();
        }
        for (int i = 0; i < 30; i++) {
            Thread thread = new Thread(new ThreadTake(arrayBlockingQueue));
            thread.start();
        }

        System.out.println("====================================================");

        /**
         * LinkedBlockingQueue 底层基于单向链表实现的阻塞队列，
         * 可以当做无界队列也可以当做有界队列来使用，同样满足FIFO的特性，
         * 与ArrayBlockingQueue 相比起来具有更高的吞吐量，为了防止 LinkedBlockingQueue 容量迅速增，
         * 损耗大量内存。通常在创建LinkedBlockingQueue 对象时，会指定其大小，如果未指定，容量等于Integer.MAX_VALUE。
         */
        //LinkedBlockingQueue可以不设置队列长度，则默认为Integer.MAX
        LinkedBlockingQueue<Integer> linkedBlockingQueue = new LinkedBlockingQueue<>();
        for (int i = 0; i < 30; i++) {
            Thread thread = new Thread(new ThreadPut(linkedBlockingQueue,"ArrayBlockingQueue",i));
            thread.start();
        }
        for (int i = 0; i < 30; i++) {
            Thread thread = new Thread(new ThreadTake(linkedBlockingQueue));
            thread.start();
        }
        System.out.println("====================================================");

        /**
         * PriorityBlockingQueue 是一个支持优先级的无界阻塞队列。
         * 默认情况下元素采用自然顺序进行排序，也可以通过自定义类实现 compareTo() 方法来指定元素排序规则，或者初始化时通过构造器参数 Comparator 来指定排序规则。
         *
         * PriorityBlockingQueue 并发控制采用的是 ReentrantLock，
         * 队列为无界队列（ArrayBlockingQueue 是有界队列，LinkedBlockingQueue 也可以通过在构造函数中传入 capacity 指定队列最大的容量，
         * 但是 PriorityBlockingQueue 只能指定初始的队列大小，后面插入元素的时候，如果空间不够的话会自动扩容）。
         *
         * 简单地说，它就是 PriorityQueue 的线程安全版本。
         * 不可以插入 null 值，同时，插入队列的对象必须是可比较大小的（comparable），
         * 否则报 ClassCastException 异常。它的插入操作 put 方法不会 block，因为它是无界队列（take 方法在队列为空的时候会阻塞）。
         */
        PriorityBlockingQueue priorityBlockingQueue = new PriorityBlockingQueue();
        for (int i = 0; i < 30; i++) {
            Thread thread = new Thread(new ThreadPut(priorityBlockingQueue,"SynchronousQueue",i));
            thread.start();
        }
        for (int i = 0; i < 30; i++) {
            Thread thread = new Thread(new ThreadTake(priorityBlockingQueue));
            thread.start();
        }
        System.out.println("====================================================");

        /**
         * SynchronousQueue是一个没有数据缓冲的BlockingQueue，生产者线程对其的插入操作put必须等待消费者的移除操作take，反过来也一样。
         */
        SynchronousQueue<Integer> synchronousQueue = new SynchronousQueue<>();
        for (int i = 0; i < 30; i++) {
            Thread thread = new Thread(new ThreadPut(synchronousQueue,"SynchronousQueue",i));
            thread.start();
        }
        for (int i = 0; i < 30; i++) {
            Thread thread = new Thread(new ThreadTake(synchronousQueue));
            thread.start();
        }

    }
}
class ThreadPut implements Runnable{
    private BlockingQueue blockingQueue;
    private String name;
    private int a;
    public ThreadPut(BlockingQueue blockingQueue,String name,int a){
        this.blockingQueue = blockingQueue;
        this.name = name;
        this.a = a;
    }
    @Override
    public void run() {
        try {
            blockingQueue.put(a);
            System.out.println(name+"--put--size--"+blockingQueue.size());
        }catch (Exception e){

        }
    }
}

class ThreadTake implements Runnable{
    private BlockingQueue blockingQueue;
    public ThreadTake (BlockingQueue blockingQueue){
        this.blockingQueue = blockingQueue;
    }
    @Override
    public void run() {
        try {
            System.out.println("take--size--"+blockingQueue.size());
            System.out.println("take=="+blockingQueue.take());
        }catch (Exception e){

        }
    }
}
